package sql

import (
	"context"
	"database/sql"
	"fmt"
	"reflect"

	"github.com/go-sql-driver/mysql"
	"github.com/jmoiron/sqlx"
	"github.com/opentracing/opentracing-go"
	"github.com/opentracing/opentracing-go/ext"
	"github.com/prometheus/client_golang/prometheus"

	"go-papers/pkg/lognew"
)

var (
	mysqlErrorCount = prometheus.NewCounterVec(
		prometheus.CounterOpts{Name: "midu_mysql_error_count", Help: "mysql error count"},
		[]string{"name"})
)

func init() {
	//prometheus.MustRegister(mysqlErrorCount)
}

type Client struct {
	name     string
	masterDB *DB
	slaveDB  *DB
	master   *Client
	host     string
	dns      string
	driver   string
	db       string
}

type DB struct {
	*sqlx.DB
	connConfig     *mysql.Config
	maxOpenConns   int32
	maxIdleConns   int32
	maxLifeTimeout float64
}

// Master return *DB instance direct use master conn
// use this *DB instance only when you have some reason need to get result without any delay.
func (c *Client) Master() *Client {
	return c.master
}

//MasterHost
func (c *Client) MasterHost() string {
	return c.host
}

func (c *Client) MasterDNS() string {
	return c.dns
}

//MasterDriver
func (c *Client) MasterDriver() string {
	if c.driver == "" {
		return "mysql"
	}
	return c.driver
}

//MasterDB
func (c *Client) MasterDB() string {
	return c.db
}

func (c *Client) Select(ctx context.Context, dest interface{}, query string, args ...interface{}) (err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_select", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, c.slaveDB.connConfig, query, args...)
	err = c.slaveDB.Select(dest, query, args...)
	handleErr(ctx, c.name, err, span, c.slaveDB.connConfig, query, args...)
	return
}

func setCommonSpan(span opentracing.Span, config *mysql.Config, query string, args ...interface{}) {

	ext.SpanKindRPCClient.Set(span)
	ext.DBType.Set(span, "mysql")
	ext.PeerService.Set(span, "mysql")
	ext.PeerHostname.Set(span, config.Addr)
	ext.DBUser.Set(span, config.User)
	ext.DBInstance.Set(span, config.DBName)
	ext.DBStatement.Set(span, toSql(query, args...))
}

func handleErr(ctx context.Context, name string, err error, span opentracing.Span, config *mysql.Config, query string, args ...interface{}) {
	if err == nil || err == sql.ErrNoRows {
		return
	}

	// 设置log错误记录
	logContext := lognew.Fields{
		"name":       name,
		"addr":       config.Addr,
		"DBInstance": config.DBName,
		"DBUser":     config.User,
		"sql":        toSql(query, args...),
		"err":        err.Error(),
	}

	// 记录错误
	lognew.Error(ctx, "mysql_error", logContext)

	// 记录监控
	mysqlErrorCount.WithLabelValues(name).Inc()

	// 设置trace错误
	ext.Error.Set(span, true)
	span.LogKV("event", "error", "message", err.Error())

}

func toSql(query string, args ...interface{}) string {
	if len(args) == 0 {
		return query
	}
	// 如果参数长度为1并且是结构体或结构体指针，
	if len(args) == 1 {
		arg := args[0]
		argType := reflect.TypeOf(arg)
		if argType.Kind() == reflect.Ptr {
			argType = argType.Elem()
		}
		if argType.Kind() == reflect.Struct {
			query, arg, err := sqlx.Named(query, arg)
			if err != nil {
				return fmt.Sprintf("query: %s, args: %v, err: %s", query, arg, err.Error())
			}
			return fmt.Sprintf("query: %s, args: %v", query, arg)
		}
	}
	return fmt.Sprintf("query: %s, args: %v", query, args)
}

func (c *Client) SelectOne(ctx context.Context, dest interface{}, query string, args ...interface{}) (err error) {
	// query 语句后面增加 limit 1 限制
	query += " limit 1"

	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_selectone", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, c.slaveDB.connConfig, query, args...)
	err = c.slaveDB.Get(dest, query, args...)
	handleErr(ctx, c.name, err, span, c.slaveDB.connConfig, query, args...)
	return
}

func (c *Client) Exec(ctx context.Context, query string, args ...interface{}) (res sql.Result, err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_exec", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, c.masterDB.connConfig, query, args...)

	res, err = c.masterDB.Exec(query, args...)
	handleErr(ctx, c.name, err, span, c.masterDB.connConfig, query, args...)
	return
}

func (c *Client) NamedExec(ctx context.Context, query string, arg interface{}) (res sql.Result, err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_namedexec", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, c.masterDB.connConfig, query, arg)

	res, err = c.masterDB.NamedExec(query, arg)
	handleErr(ctx, c.name, err, span, c.masterDB.connConfig, query, arg)
	return
}

type Tx struct {
	*sqlx.Tx
	name       string
	connConfig *mysql.Config
}

func (c *Client) Begin(ctx context.Context) (tx *Tx, err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_begin", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, c.masterDB.connConfig, "begin")
	tx = &Tx{name: c.name, connConfig: c.masterDB.connConfig}
	tx.Tx, err = c.masterDB.Beginx()
	handleErr(ctx, c.name, err, span, c.masterDB.connConfig, "begin")
	return
}

func (tx *Tx) Select(ctx context.Context, dest interface{}, query string, args ...interface{}) (err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_tx_select", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, tx.connConfig, query, args...)
	err = tx.Tx.Select(dest, query, args...)
	handleErr(ctx, tx.name, err, span, tx.connConfig, query, args...)
	return
}

func (tx *Tx) SelectOne(ctx context.Context, dest interface{}, query string, args ...interface{}) (err error) {
	// query 语句后面增加 limit 1 限制
	query += " limit 1"

	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_tx_selectone", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, tx.connConfig, query, args...)
	err = tx.Tx.Get(dest, query, args...)
	handleErr(ctx, tx.name, err, span, tx.connConfig, query, args...)
	return
}

func (tx *Tx) Exec(ctx context.Context, query string, args ...interface{}) (res sql.Result, err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_tx_exec", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, tx.connConfig, query, args...)
	res, err = tx.Tx.Exec(query, args...)
	handleErr(ctx, tx.name, err, span, tx.connConfig, query, args...)
	return
}

func (tx *Tx) Commit(ctx context.Context) (err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_tx_commit", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, tx.connConfig, "commit")
	err = tx.Tx.Commit()
	handleErr(ctx, tx.name, err, span, tx.connConfig, "commit")
	return
}

func (tx *Tx) Rollback(ctx context.Context) (err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_tx_rollback", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, tx.connConfig, "rollback")
	err = tx.Tx.Rollback()
	handleErr(ctx, tx.name, err, span, tx.connConfig, "rollback")
	return
}

func (tx *Tx) NamedExec(ctx context.Context, query string, arg interface{}) (res sql.Result, err error) {
	tc := Trace(ctx)
	span := tc.tracer.StartSpan("mysql_tx_namedexec", opentracing.ChildOf(tc.spanCtx))
	defer span.Finish()
	setCommonSpan(span, tx.connConfig, query, arg)

	res, err = tx.Tx.NamedExec(query, arg)
	handleErr(ctx, tx.name, err, span, tx.connConfig, query, arg)
	return
}
